package Test;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;
import scala.Serializable;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Random;

public class TestSparkSortByKey {
    public static void main(String[] args) {
        // SparkSortByKey 二次排序
        SparkConf sparkConf = new SparkConf().setAppName("TestSparkSortByKey").setMaster("local");
        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
        List<Integer> data = Arrays.asList(5, 1, 1, 4, 4, 2, 2, 5, 5);

        JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data);

        final Random random = new Random(100);

        JavaPairRDD<String, Integer> javaPairRDD = javaRDD.mapToPair(new PairFunction<Integer, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Integer integer) throws Exception {
                return new Tuple2<String, Integer>(Integer.toString(integer) + "," + random.nextInt(10),random.nextInt(10));
            }
        });

        JavaPairRDD<String,Integer> sortByKeyRDD = javaPairRDD.sortByKey(new Comp());
        System.out.println(sortByKeyRDD.collect());
    }

}
class Comp implements Comparator<String>,Serializable{
    @Override
    public int compare(String o1, String o2) {
          String []o1s = o1.split(",");
          String []o2s = o2.split(",");
          if(o1s[0].compareTo(o2s[0]) == 0)
              return o1s[1].compareTo(o2s[1]);
           else
                return -o1s[0].compareTo(o2s[0]);
  }
}

